package com.zengqingfa.rocketmq.sender;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.zengqingfa.rocketmq.producer.MqDemo01Producer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 *
 * @fileName: MqDemo01Producer
 * @author: zengqf3
 * @date: 2021-4-2 22:47
 * @description: 生产者
 */
@Component
@Slf4j
public class MqDemo01Sender {

    @Value("${rocketmq.topic.demo01}")
    private String topic;

    @Value("${rocketmq.tag.demo01}")
    private String tag;

    @Autowired
    MqDemo01Producer mqDemo01Producer;

    /**
     * 消息发送
     * @param msgs
     * @throws Exception
     */
    public void send(List<Message> msgs) throws Exception {
        for (Message msg : msgs) {
            String body = JSON.toJSONString(msg);
            Message sendMsg = new Message(topic, tag, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
            log.info("消息:topic={},tags={},key={},body={}", msg.getTopic(), msg.getTags(), msg.getKeys(), body);
            SendResult sendResult = mqDemo01Producer.getProducer().send(sendMsg);
            log.info("发送结果 status={},topic={},queueId={},msgId={}", sendResult.getSendStatus(),
                    sendResult.getMessageQueue().getTopic(), sendResult.getMessageQueue().getQueueId(),
                    sendResult.getMsgId());
        }
    }
}
